package com.example.mq.mqServer.center;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.mqServer.core.MSGQueue;
import com.example.mq.mqServer.VirtualHost;
import com.example.mq.mqServer.core.message.Message;
import com.example.mq.mqServer.exception.MqException;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * @BelongsProject: 0.mq
 * @BelongsPackage: com.example.mq.mqServer.core
 * @Author : 游在海里的鱼
 * @CreateTime : 2023-12-21  14:00
 * @Version : 1.0
 * @Description : 消费管理者
 */
@Slf4j
public class ConsumerCenter {
    //虚拟机
    private VirtualHost virtualHost;
    //虚拟机的内存中心
    private MemoryDataCenter memoryDataCenter;
    //线程池
    private ExecutorService workPoll = Executors.newWorkStealingPool(4);
    //令牌的阻塞队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    //扫描线程
    private Thread scannerThread=null;
    //扫描线程的运行状态
    private volatile boolean running;

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 构造方法
     * @Describe: TODO
     * @modifier: [public] [构造方法]
    **/
    public ConsumerCenter(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        this.memoryDataCenter = virtualHost.getMemoryDataCenter();
        scannerThread = new Thread(()->{
            while (running){
                try {
                    //1.获取令牌
                    String queueName = tokenQueue.take();
                    //2.获取队列,根据令牌
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        new MqException("[扫描线程] [令牌"+queueName+"]所指队列不存在").printStackTrace();
                        break;
                    }
                    //3.消费此队列中的消息
                    synchronized (queue){
                        consumerMessage(queue);
                    }
                } catch (InterruptedException e) {
                    log.error("[扫描线程] [队列令牌]从[令牌队列]不能正常取出");
                    e.printStackTrace();
                }
            }
        });
        //设置为后台线程,当前台线程全部结束后,后台线程关闭
        scannerThread.setDaemon(true);
        running = true;
        scannerThread.start();
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 关闭扫描线程
     * @Describe: TODO
     * @modifier: [public] [实例方法]
     **/
    public void close(){
        running = false;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 添加消费者为队列，消费者订阅队列
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public void addConsumer(String consumerTag, MSGQueue queue, boolean autoAck, Consumer consumer){
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queue.getName(),autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            for(int i=0; i<memoryDataCenter.sizeMessage(queue); i++){
                consumerMessage(queue);
            }
        }
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 消费消息
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private void consumerMessage(MSGQueue queue){
        //1.指定消费者,按照轮询的方式
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if(consumerEnv == null){
            return;
        }
        //2.获取消息
        Message message = memoryDataCenter.pollMessage(queue);
        if(message == null){
            log.error("[consumerMessage] [队列"+queue.getName()+"]中没有[消息]");
            return;
        }
        //3.将消费者的回调函数加入线程池
        workPoll.submit(()->{
            //添加到未回应消息集合
            memoryDataCenter.addUnAck(queue.getName(),message);
            //执行消费者的回调函数
            consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag()
                    ,message.getBasicProperties(), message.getBody());
            //如果是手动应答,则暂不处理
            //如果是自动应答,则在此应答消息
            if(consumerEnv.isAutoACK()){
                virtualHost.basicAck(queue.getName(),message.getId());
            }
            log.debug("[扫描线程] [队列"+queue.getName()+"]的[消息"+message.getId()+"]成功消费");
        });
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 通知消费者来取消息
     * @Describe: TODO
     * @modifier: [public] [实例方法]
     **/
    public void notifyConsume(String queueName){
        try {
            tokenQueue.put(queueName);
        } catch (InterruptedException e) {
            log.error("[notifyConsume] [队列令牌]添加[令牌队列]失败");
            e.printStackTrace();
        }
    }
}
